package cn.liangxxhu.rocketmq.core;

import cn.liangxxhu.rocketmq.core.param.MessageWrapper;
import cn.liangxxhu.rocketmq.core.param.RocketmqFailedMessage;
import cn.liangxxhu.rocketmq.toolkit.TraceIdUtil;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.TypeReference;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.util.Assert;

import java.nio.charset.StandardCharsets;
import java.time.LocalDateTime;

/**
 * RocketMQ 消费消息抽象类，统一封装消费主流程（反序列化、链路追踪、重试与落库）。
 *
 * @param <T> 业务消息体类型
 */
@Slf4j
@RequiredArgsConstructor
public abstract class AbstractCommonConsumerTemplate<T> {

    /**
     * 消费失败消息落库
     */
    private final FailedMessageDao failedMessageDao;

    /**
     * 默认最大重试次数。
     * 说明：基于 RocketMQ 的 reconsumeTimes 统计，当 reconsumeTimes 大于该值时判定为超限。
     */
    private static final int DEFAULT_MAX_RETRY_COUNT = 3;

    /**
     * 业务消费逻辑处理（子类必须实现）。
     *
     * @param message 消息包装器（含业务数据和元信息）
     */
    protected abstract void handleMessage(MessageWrapper<T> message);

    /**
     * Fastjson 反序列化类型引用（用于反序列化包装器）。
     *
     * @return 带泛型的 TypeReference
     */
    protected abstract TypeReference<MessageWrapper<T>> getMessageWrapperType();

    /**
     * 最大重试次数（子类可覆盖自定义）。
     *
     * @return 最大重试次数
     */
    protected int getMaxRetryCount() {
        return DEFAULT_MAX_RETRY_COUNT;
    }

    /**
     * 超过最大重试次数后的处理逻辑：消息落库 + 告警兜底。
     *
     * @param message 消息包装器
     */
    protected void handleMaxRetriesExceeded(MessageWrapper<T> message) {
        // 日志上下文增强，便于追踪
        String logContext = buildLogContext(message);
        log.error("消息重试超限（{}次），开始处理 | {}", getMaxRetryCount(), logContext);
        try {
            // 构建失败消息实体
            RocketmqFailedMessage failedMsg = buildFailedMessage(message);
            // 执行落库
            failedMessageDao.save(failedMsg);
            log.info("消息落库成功 | {}", logContext);
        } catch (Exception e) {
            // 落库失败时增强日志，便于排查
            log.error("消息落库失败 | {} | 失败原因: {}", logContext, e.getMessage(), e);
            // 可扩展：发送告警通知（如钉钉/邮件）
            sendFallbackAlert(message, e);
            // 抛出异常标记消费失败（根据业务决定是否继续重试）
            throw new RuntimeException("消息重试超限且落库失败", e);
        }
    }

    /**
     * 构建日志上下文（统一格式，便于检索）。
     */
    private String buildLogContext(MessageWrapper<T> message) {
        return String.format(
                "UUID: %s | Keys: %s | 事件: %s | 主题: %s | 重试次数: %d",
                message.getUuid(),
                message.getKeys(),
                message.getEventName(),
                message.getTopic(),
                message.getReconsumeTimes()
        );
    }

    /**
     * 构建失败消息实体（映射数据库表结构）。
     */
    private RocketmqFailedMessage buildFailedMessage(MessageWrapper<T> message) {
        RocketmqFailedMessage failedMsg = new RocketmqFailedMessage();
        failedMsg.setMessageUuid(message.getUuid());
        failedMsg.setTopic(message.getTopic());
        failedMsg.setTag(message.getTag()); // 允许为null
        failedMsg.setKeys(message.getKeys()); // 允许为null
        failedMsg.setEventName(message.getEventName()); // 允许为null
        failedMsg.setMessageBody(JSON.toJSONString(message.getMessage())); // 序列化业务消息体
        failedMsg.setReconsumeTimes(message.getReconsumeTimes());
        failedMsg.setFailReason(String.format("超过最大重试次数阈值: %d次", getMaxRetryCount()));
        failedMsg.setSendTimestamp(message.getTimestamp()); // 原始发送时间
        failedMsg.setCreateTime(LocalDateTime.now()); // 落库时间
        return failedMsg;
    }

    /**
     * 落库失败的降级告警（可子类重写扩展）。
     */
    protected void sendFallbackAlert(MessageWrapper<T> message, Exception e) {
        // 示例：打印紧急日志（实际可对接告警系统）
        log.error("[紧急] 消息落库失败可能导致数据丢失 | UUID: {} | 主题: {}", message.getUuid(), message.getTopic());
    }

    /**
     * 消费主逻辑入口。
     */
    public void dispatchMessage(MessageWrapper<T> message) {
        // 入参校验，提前暴露问题
        Assert.notNull(message, "消息包装器不能为空");
        String logContext = buildLogContext(message);
        try {
            // 绑定链路追踪ID
            TraceIdUtil.putTraceId(message.getKeys());
            log.info("开始消费消息 | {}", logContext);
            // 判断是否超过最大重试次数
            if (message.getReconsumeTimes() > getMaxRetryCount()) {
                handleMaxRetriesExceeded(message);
                return;
            }
            // 执行业务消费逻辑
            long start = System.currentTimeMillis();
            handleMessage(message);
            log.info("消息消费成功 | {} | 耗时: {}ms", logContext, System.currentTimeMillis() - start);
        } catch (Exception e) {
            // 消费失败日志增强，包含耗时和异常信息
            log.error("消息消费失败 | {} | 失败原因: {}", logContext, e.getMessage(), e);
            // 抛出异常触发RocketMQ重试（框架会自动累加reconsumeTimes）
            throw new RuntimeException("消息消费失败，触发重试", e);
        } finally {
            // 清理链路追踪ID，避免线程复用导致的污染
            TraceIdUtil.clear();
        }
    }

    /**
     * 从 RocketMQ 原始消息构建包装器（补充元信息）。
     */
    protected MessageWrapper<T> buildWrapperFromMessageExt(MessageExt messageExt) {
        // 校验原始消息
        Assert.notNull(messageExt, "RocketMQ原始消息不能为空");
        Assert.notNull(messageExt.getBody(), "消息体字节数组不能为空");
        try {
            // 反序列化消息体
            String body = new String(messageExt.getBody(), StandardCharsets.UTF_8);
            MessageWrapper<T> wrapper = JSON.parseObject(body, getMessageWrapperType());
            // 补充元信息（从原始消息提取）
            wrapper.setReconsumeTimes(messageExt.getReconsumeTimes());
            return wrapper;
        } catch (Exception e) {
            log.error("构建消息包装器失败 | MsgId: {} | 消息体: {}", messageExt.getMsgId(), new String(messageExt.getBody()), e);
            throw new RuntimeException("消息反序列化失败", e);
        }
    }
}